package com.hzy.service.document;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.CompositeFuture;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.web.client.WebClient;
import io.vertx.ext.web.client.WebClientOptions;
import io.vertx.pgclient.PgConnectOptions;
import io.vertx.pgclient.PgPool;
import io.vertx.sqlclient.PoolOptions;
import io.vertx.sqlclient.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.UUID;

public class PgDocumentVerticle extends AbstractVerticle implements DocumentVerticle {

    private static final Logger LOGGER = LoggerFactory.getLogger(PgDocumentVerticle.class);


    private PgPool pgPool;
    private WebClient esWebClient;
    private String esIndexName;


    public void start(Promise<Void> startPromise) {
        JsonObject config = (JsonObject) vertx.sharedData().getLocalMap("config").get("appConfig");

        // 数据库配置
        JsonObject dbConfig = config.getJsonObject("database");
        PgConnectOptions connectOptions = new PgConnectOptions()
                .setPort(dbConfig.getInteger("port"))
                .setHost(dbConfig.getString("host"))
                .setDatabase(dbConfig.getString("database"))
                .setUser(dbConfig.getString("username"))
                .setPassword(dbConfig.getString("password"));

        PoolOptions poolOptions = new PoolOptions()
                .setMaxSize(dbConfig.getInteger("poolSize", 5));

        pgPool = PgPool.pool(vertx, connectOptions, poolOptions);

        initializeDatabase(pgPool).future()
                .onSuccess(v -> {
                    LOGGER.info("PostgreSQL database initialized successfully.");

                    registerServiceConsumer(vertx, ADD_DOCUMENT_ADDRESS, this::addDocument);
                    registerServiceConsumer(vertx, GET_DOCUMENT_BY_ID_ADDRESS, this::getDocumentById);
                    registerServiceConsumer(vertx, SEARCH_DOCUMENTS_ADDRESS, this::searchDocuments);

                    startPromise.complete(); // 成功完成 Verticle 部署
                })
                .onFailure(throwable -> {
                    LOGGER.error("Failed to initialize PostgreSQL database: {}", throwable.getMessage());
                    startPromise.fail(throwable);
                });


        // Elasticsearch WebClient 配置
        JsonObject esConfig = config.getJsonObject("elasticsearch");
        if (esConfig != null) {
            WebClientOptions esOptions = new WebClientOptions()
                    .setDefaultHost(esConfig.getString("host", "localhost"))
                    .setDefaultPort(esConfig.getInteger("port", 9200));
            esWebClient = WebClient.create(vertx, esOptions);
            esIndexName = esConfig.getString("indexName", "documents");
        } else {
            LOGGER.warn("Elasticsearch configuration not found. Document search via ES will be unavailable.");
        }
    }


    private Promise<Void> initializeDatabase(PgPool pool) {
        Promise<Void> promise = Promise.promise();
        pool.query("CREATE TABLE IF NOT EXISTS documents (id TEXT PRIMARY KEY, title TEXT NOT NULL, content TEXT)")
                .execute()
                .onSuccess(v -> promise.complete())
                .onFailure(promise::fail);
        return promise;
    }


    public Promise<String> addDocument(JsonObject document) {
        Promise<String> promise = Promise.promise();
        String id = document.getString("id", UUID.randomUUID().toString()); // 使用现有ID或生成新ID
        String title = document.getString("title");
        String content = document.getString("content");

        if (title == null || content == null) {
            promise.fail("Title and content are required for a document.");
            return promise;
        }

        pgPool.preparedQuery("INSERT INTO documents (id, title, content) VALUES ($1, $2, $3) RETURNING id")
                .execute(io.vertx.sqlclient.Tuple.of(id, title, content))
                .onSuccess(rowSet -> {
                    if (rowSet.iterator().hasNext()) {
                        String returnedId = rowSet.iterator().next().getString("id");
                        // 如果有 Elasticsearch 集成，同时添加到 Elasticsearch
                        if (esWebClient != null && esIndexName != null) {
                            JsonObject esDoc = new JsonObject()
                                    .put("id", returnedId)
                                    .put("title", title)
                                    .put("content", content);
                            esWebClient.post("/" + esIndexName + "/_doc/" + returnedId)
                                    .sendJsonObject(esDoc)
                                    .onSuccess(esResponse -> {
                                        if (esResponse.statusCode() == 200 || esResponse.statusCode() == 201) {
                                            LOGGER.info("Document {} added to PostgreSQL and Elasticsearch.", returnedId);
                                            promise.complete(returnedId);
                                        } else {
                                            String errorMsg = "Failed to add document to Elasticsearch: " + esResponse.bodyAsString();
                                            LOGGER.error(errorMsg);
                                            // 即使ES失败，PostgreSQL也成功了，所以考虑完成promise
                                            promise.complete(returnedId); // 或者promise.fail(errorMsg); 取决于业务需求
                                        }
                                    })
                                    .onFailure(throwable -> {
                                        LOGGER.error("Failed to connect to Elasticsearch to add document {}: {}", returnedId, throwable.getMessage());
                                        // 即使ES连接失败，PostgreSQL也成功了
                                        promise.complete(returnedId); // 或者promise.fail(throwable);
                                    });
                        } else {
                            // 没有 Elasticsearch 配置，只完成 PostgreSQL 写入
                            LOGGER.info("Document {} added to PostgreSQL only (Elasticsearch not configured).", returnedId);
                            promise.complete(returnedId);
                        }
                    } else {
                        promise.fail("Failed to add document, no ID returned.");
                    }
                })
                .onFailure(promise::fail);
        return promise;
    }


    public Promise<JsonObject> getDocumentById(String docId) {
        Promise<JsonObject> promise = Promise.promise();
        pgPool.preparedQuery("SELECT id, title, content FROM documents WHERE id = $1")
                .execute(io.vertx.sqlclient.Tuple.of(docId))
                .onSuccess(rowSet -> {
                    if (rowSet.iterator().hasNext()) {
                        Row row = rowSet.iterator().next();
                        JsonObject document = new JsonObject()
                                .put("id", row.getString("id"))
                                .put("title", row.getString("title"))
                                .put("content", row.getString("content"));
                        promise.complete(document);
                    } else {
                        promise.complete(null); // Document not found
                    }
                })
                .onFailure(promise::fail);
        return promise;
    }


    public Promise<JsonArray> searchDocuments(String query) {
        Promise<JsonArray> promise = Promise.promise();
        JsonArray retrievedContext = new JsonArray();

        if (esWebClient == null || esIndexName == null) {
            LOGGER.warn("Elasticsearch is not configured. Cannot perform document search.");
            promise.fail("Elasticsearch is not configured for document search.");
            return promise;
        }

        JsonObject searchQuery = new JsonObject()
                .put("query", new JsonObject()
                        .put("multi_match", new JsonObject()
                                .put("query", query)
                                .put("fields", new JsonArray().add("title").add("content"))));

        esWebClient.post("/" + esIndexName + "/_search")
                .sendJsonObject(searchQuery)
                .onSuccess(response -> {
                    if (response.statusCode() == 200) {
                        JsonObject responseBody = response.bodyAsJsonObject();
                        JsonArray hits = responseBody.getJsonObject("hits").getJsonArray("hits");
                        for (int i = 0; i < hits.size(); i++) {
                            // 假设我们只关心 content 字段作为上下文
                            retrievedContext.add(hits.getJsonObject(i).getJsonObject("_source").getString("content"));
                        }
                        LOGGER.info("Elasticsearch search for query '{}' returned {} hits.", query, hits.size());
                        promise.complete(retrievedContext);
                    } else {
                        String errorMsg = "Elasticsearch search failed: " + response.bodyAsString();
                        LOGGER.error(errorMsg);
                        promise.fail(errorMsg);
                    }
                })
                .onFailure(throwable -> {
                    LOGGER.error("Failed to connect to Elasticsearch for search: {}", throwable.getMessage());
                    promise.fail("Failed to connect to Elasticsearch for search: " + throwable.getMessage());
                });
        return promise;
    }


    public void stop(Promise<Void> stopPromise) {
        List<Future<Void>> closeFutures = new ArrayList<>();

        if (pgPool != null) {
            closeFutures.add(pgPool.close()
                    .onComplete(ar -> {
                        if (ar.succeeded()) {
                            LOGGER.info("PostgreSQL connection pool closed.");
                        } else {
                            LOGGER.error("Failed to close PostgreSQL connection pool", ar.cause());
                        }
                    }));
        }

        if (esWebClient != null) {
            try {
                esWebClient.close();
                LOGGER.info("Elasticsearch WebClient closed.");
                closeFutures.add(Future.succeededFuture()); // WebClient.close() 是同步的，包装为成功的 Future
            } catch (Exception e) {
                LOGGER.error("Failed to close Elasticsearch WebClient", e);
                closeFutures.add(Future.failedFuture(e)); // 如果关闭过程中出现异常，则包装为失败的 Future
            }
        }

        // 使用 CompositeFuture 等待所有关闭操作完成
        if (closeFutures.isEmpty()) {
            stopPromise.complete(); // 如果没有需要关闭的资源，直接完成
        } else {
            CompositeFuture.all(Collections.unmodifiableList(closeFutures))
                    .onComplete(ar -> {
                        if (ar.succeeded()) {
                            LOGGER.info("All resources closed successfully.");
                            stopPromise.complete();
                        } else {
                            LOGGER.error("One or more resources failed to close during stop: {}", ar.cause().getMessage(), ar.cause());
                            stopPromise.fail(ar.cause());
                        }
                    });
        }
    }
}